#include <cstdlib>
#include <cstring>
#include <unistd.h>
#include "master.h"
#include "debug.h"

int Master::parse_conf()
{
	debug(SECTION_MASTER, 10, "---> parse_conf\n");
	int ret = 0;
	FILE *fp;
	char opt_name[100];
	char opt_val[100];

	if ((fp = fopen(conf.c_str(), "r")) != NULL) {
		while (fscanf(fp, "%s\t=\t%s", opt_name, opt_val) != EOF) {
			debug(SECTION_MASTER, 10, "CONFIG : %s = %s\n", opt_name, opt_val);
			if (!strcmp(opt_name, "master_id2uri")) {
				char id_[5];
				char uri_[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id_, uri_);
				//id = atoi(id_);
				uri = strdup(uri_);
			} else if (!strcmp(opt_name, "cs_id2uri")) {
				char id_[5];
				char uri_[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id_, uri_);
				uint64_t cs_id = atoi(id_);
				chunkservers[cs_id].id = cs_id;
				chunkservers[cs_id].uri = strdup(uri_);
				chunkservers[cs_id].state = DOWN;
				heartbeatmap[cs_id].chunkserver = cs_id;
				heartbeatmap[cs_id].last_tid = 0;
				heartbeatmap[cs_id].last_heartbeat = 0;
			} else if (!strcmp(opt_name, "heartbeat_interval")) {
				heartbeat_interval = atoi(opt_val);
			} else {
				debug(SECTION_MASTER, 10, "Unrecognized config option : %s\n", opt_name);
			}
		}
		fclose(fp);
	} else {
		ret = -1;
		debug(SECTION_MASTER, 10, "Failed to open conf file : %s", conf.c_str());
		return ret;
	}
	return ret;
}

void Master::chunk_op_service(void *self_, void *ex_, void *master_)
{
	debug(SECTION_MASTER, 10, "---> chunk_op_service\n");
	RdmaRpc *rpc = static_cast<RdmaRpc *>(self_);
	RpcExecution *ex = static_cast<RpcExecution *>(ex_);
	Master *master = static_cast<Master *>(master_);

	RpcMsg *rpc_req = ex->req_msg;
	RpcMsg *rpc_resp = ex->resp_msg;
	struct dss_msg *msg = reinterpret_cast<struct dss_msg*>(rpc_req->data);
	struct dss_msg *msg_resp = reinterpret_cast<struct dss_msg*>(rpc_resp->data);

	struct ChunkOp *op = static_cast<struct ChunkOp*>(master->rpc_client->AllocPiece());
	memset(op, 0, sizeof(struct ChunkOp));
	op->master = master;
	op->rpc_ex = ex;
	op->dss_request = msg;
	op->dss_response = msg_resp;
	switch (msg->msg_type)
	{
	case CHUNK_OP_CREATE:
		master->handle_op_create(op);
		break;
	case CHUNK_OP_LIST:
		master->handle_op_list(op);
		break;
	default:
		//TODO unexpect message
		debug(SECTION_MASTER, 10, "unexpect message\n");
	}
	debug(SECTION_MASTER, 10, "<--- chunk_op_service\n");
	return;
}

void Master::sys_status_service(void *self_, void *ex_, void *master_)
{
	debug(SECTION_MASTER, 10, "---> sys_status_service\n");

	debug(SECTION_MASTER, 10, "<--- sys_status_service\n");
}

void Master::heartbeat_service(void *self_, void *ex_, void *master_)
{
	debug(SECTION_MASTER, 10, "---> heartbeat_service\n");
	RdmaRpc *rpc = static_cast<RdmaRpc *>(self_);
	RpcExecution *ex = static_cast<RpcExecution *>(ex_);
	Master *master = static_cast<Master *>(master_);

	RpcMsg *rpc_req = ex->req_msg;
	RpcMsg *rpc_resp = ex->resp_msg;
	struct dss_msg *msg = reinterpret_cast<struct dss_msg*>(rpc_req->data);
	struct dss_msg *heart_resp = reinterpret_cast<struct dss_msg*>(rpc_resp->data);
	
	struct heartbeat_req *heart_header = (struct heartbeat_req *)msg->data;
	/* 1. update master->chunkservers; 2. update master->heartbeatmap */
	struct ChunkServer *cs = &(master->chunkservers[heart_header->cs_id]);
	struct HeartbeatInfo *heart_info = &(master->heartbeatmap[heart_header->cs_id]);

	heart_info->last_heartbeat = clock();
	heart_info->last_tid = msg->msg_id;
	cs->cluster_total = heart_header->cluster_total;
	cs->cluster_free = heart_header->cluster_free;
	cs->cluster_size = heart_header->cluster_size;
	cs->io_unit_size = heart_header->io_unit_size;

	/* set value to heartbeat_response */
	heart_resp->status_code = SUCCESS;
	heart_resp->msg_type = CS_HEARTBEAT_REPLY;
	heart_resp->msg_id = msg->msg_id;
	heart_resp->data_len = 0;

	rpc_resp->status_code = SUCCESS;
	rpc_resp->data_len = sizeof(struct dss_msg) + heart_resp->data_len;
	rpc->RpcServiceComplete(ex);
	debug(SECTION_MASTER, 10, "<--- heartbeat_service\n");
}


void Master::handle_op_create(struct ChunkOp *op)
{
	debug(SECTION_MASTER, 10, "---> handle_op_create\n");
	op->total_response = 1;
	op->cur_response = 0;
	/* to select a chunkserver to create chunk */
	int sess_id = chunkserver_select();
	if (sess_id < 0) {
		/* set value to create_response */
		op->dss_response->status_code = FAILURE;
		op->dss_response->msg_type = CHUNK_OP_CREATE_REPLY;
		op->dss_response->msg_id = op->dss_request->msg_id;
		op->dss_response->data_len = 0;
		RpcMsg *rpc_resp = op->rpc_ex->resp_msg;
		rpc_resp->status_code = SUCCESS;
		rpc_resp->data_len = sizeof(struct dss_msg) + op->dss_response->data_len;
		rpc_server->RpcServiceComplete(op->rpc_ex);
		rpc_client->FreePiece(op);
		debug(SECTION_MASTER, 10, "no accessible chunkserver!!!!!!!!!\n");
		return;
	}
	/* construct a new request to chunkserver to create chunk */
	RpcMsg *rpc_header = static_cast<RpcMsg*>(rpc_client->AllocMsg());
	rpc_client->PrepareMsg(rpc_header, S_CHUNK_IO_SERVICE);
	struct dss_msg *dss_header = reinterpret_cast<struct dss_msg*>(rpc_header->data);
	dss_header->client_id = -1;
	dss_header->msg_type = CHUNK_OP_CREATE;
	dss_header->msg_id = cur_id++;
	dss_header->data_len = 0;

	rpc_header->data_len = sizeof(struct dss_msg) + dss_header->data_len;
	rpc_client->RpcCliWriteReq(sess_id, rpc_header, create_cb, op);
	debug(SECTION_MASTER, 10, "<--- handle_op_create\n");
}

void Master::handle_op_list(struct ChunkOp *op)
{
	debug(SECTION_MASTER, 10, "---> handle_op_list\n");
	int i = 0;
	op->dss_response->msg_type = CHUNK_OP_LIST_REPLY;
	op->total_response = chunkservers.size();
	op->cur_response = 0;
	RpcMsg *rpc_header = static_cast<RpcMsg*>(rpc_client->AllocMsg());
	rpc_client->PrepareMsg(rpc_header, S_CHUNK_IO_SERVICE);
	struct dss_msg *dss_header = reinterpret_cast<struct dss_msg*>(rpc_header->data);
	dss_header->client_id = -1;
	dss_header->msg_id = cur_id++;
	dss_header->msg_type = CHUNK_OP_LIST;
	dss_header->data_len = 0;
	rpc_header->data_len = sizeof(struct dss_msg) + dss_header->data_len;
	/* to each chunkserver construct a new request to get chunk list */
	for (i = 0; i < chunkservers.size(); i++) {
		/* TODO master_list_cb */
		rpc_client->RpcCliWriteReq(chunkservers[i].session_id, rpc_header, list_cb, op);
	}
	debug(SECTION_MASTER, 10, "<--- handle_op_list\n");
}


void Master::create_cb(void *arg1, void *arg2, void *arg3)
{
	debug(SECTION_MASTER, 10, "---> create_cb\n");
	RdmaRpc *rpc = static_cast<RdmaRpc *>(arg1);
	RpcExecution *ex = static_cast<RpcExecution *>(arg2);
  struct ChunkOp *op = static_cast<struct ChunkOp *>(arg3);
	RpcMsg *rpc_resp = ex->resp_msg;
  struct dss_msg *dss_resp = (struct dss_msg *)rpc_resp->data;
	Master *master = op->master;
	op->cur_response++;
	if (op->cur_response == op->total_response) {
    memcpy(op->dss_response, dss_resp, rpc_resp->data_len);
		struct chunk_resp *create_resp = (struct chunk_resp *)op->dss_response->data;
    debug(SECTION_MASTER, 10, "chunkid :[%d, %d]\n", create_resp->id.cs_id, create_resp->id.blob_id);
		RpcMsg *srpc_resp = op->rpc_ex->resp_msg;
		srpc_resp->data_len = sizeof(struct dss_msg) + sizeof(struct chunk_resp);
		master->rpc_client->free_rpc_request(ex);
		master->rpc_server->RpcServiceComplete(ex);
		master->rpc_client->FreePiece(op);
	}
	debug(SECTION_MASTER, 10, "<--- create_cb\n");
}


void Master::list_cb(void *arg1, void *arg2, void *arg3)
{
	debug(SECTION_MASTER, 10, "---> list_cb\n");
	RdmaRpc *rpc = static_cast<RdmaRpc *>(arg1);
	RpcExecution *ex = static_cast<RpcExecution *>(arg2);
  struct ChunkOp *op = static_cast<struct ChunkOp *>(arg3);
	RpcMsg *rpc_resp = ex->resp_msg;
  struct dss_msg *dss_resp = (struct dss_msg *)rpc_resp->data;
	Master *master = op->master;
	op->cur_response++;
	if (op->cur_response == op->total_response) {
  //TODO 
  }
  debug(SECTION_MASTER, 10, "<--- list_cb\n");
}
void *Master::heartbeat_main(void *arg)
{
	debug(SECTION_MASTER, 10, "---> heartbeat_main\n");

	Master *master = (Master *)arg;
	int i = 0;
	void *ret = NULL;
	struct ChunkServer *cs = NULL;
	while (master->state != EXIT) {
		if (master->state == ACTIVE) {
			/* check every chunkserver's heartbeat and update it's state */
			for (i = 0; i < master->chunkservers.size(); i++) {
				debug(SECTION_MASTER, 10, "cs:%d, last_heartbeat:%d\n", i, master->heartbeatmap[i].last_heartbeat);
				clock_t now = clock();
				uint64_t last_interval = (uint64_t)((float)(now - master->heartbeatmap[i].last_heartbeat) / CLOCKS_PER_SEC);
				debug(SECTION_MASTER, 10, "now : %d\n", now);
				debug(SECTION_MASTER, 10, "last_interval : %d\n", last_interval);
				if(last_interval > master->heartbeat_interval) {
					master->chunkservers[i].state = DOWN;
				} else if (master->chunkservers[i].state == DOWN) {
					debug(SECTION_MASTER, 10, "chunkserver: %d up\n", i);
					/* TODO : check rpc_create_session's return value!!! */
					master->chunkservers[i].session_id = master->rpc_client->ConnectSess(master->chunkservers[i].uri.c_str(), NULL, NULL);
					master->chunkservers[i].state = UP;
				}
				debug(SECTION_MASTER, 10, "cs:%d, state:%s\n", i, master->chunkservers[i].state ? "UP" : "DOWN");
			}
		} else {
			debug(SECTION_MASTER, 10, "master's state is not active!!!\n");
		}
		debug(SECTION_MASTER, 10, "heartbeat_interval = %d\n", master->heartbeat_interval);
		sleep(master->heartbeat_interval);
	}
	debug(SECTION_MASTER, 10, "<--- heartbeat_main\n");
	return ret;
}

void Master::heartbeat_manager()
{
	debug(SECTION_MASTER, 10, "---> heartbeat_manager\n");
	int i = 0;
	struct ChunkServer *cs = NULL;
	/* check every chunkserver's heartbeat and update it's state */
	for (i = 0; i < chunkservers.size(); i++) {
		debug(SECTION_MASTER, 10, "cs:%d, last_heartbeat:%d\n", i, heartbeatmap[i].last_heartbeat);
		clock_t now = clock();
		uint64_t last_interval = (uint64_t)((float)(now - heartbeatmap[i].last_heartbeat) / CLOCKS_PER_SEC);
		debug(SECTION_MASTER, 10, "now : %d\n", now);
		debug(SECTION_MASTER, 10, "last_interval : %d\n", last_interval);
		if(last_interval > heartbeat_interval) {
			chunkservers[i].state = DOWN;
		} else if (chunkservers[i].state == DOWN) {
			debug(SECTION_MASTER, 10, "chunkserver: %d up\n", i);
			/* TODO : check rpc_create_session's return value!!! */
			chunkservers[i].session_id = rpc_client->ConnectSess(chunkservers[i].uri.c_str(), NULL, NULL);
			if (chunkservers[i].session_id > 0)
				chunkservers[i].state = UP;
		}
		debug(SECTION_MASTER, 10, "cs:%d, state:%s\n", i, chunkservers[i].state ? "UP" : "DOWN");
	}
	debug(SECTION_MASTER, 10, "<--- heartbeat_manager\n");
}


int Master::chunkserver_select()
{
	debug(SECTION_MASTER, 10, "---> chunkserver_select\n");
	int cs_id = 0;
	int count = 0;
	int cs_num = chunkservers.size();
	do {
		cs_id = (cur_cs++) % cs_num;
		if (cur_cs >= cs_num)
			cur_cs = 0;
		count++;
		if (count > cs_num)
			return -1;
	} while (chunkservers[cs_id].state == DOWN);

	struct ChunkServer *cs = &(chunkservers[cs_id]);
	debug(SECTION_MASTER, 10, "<--- chunkserver_select, cs_id = %d, session_id = %d\n", cs_id, cs->session_id);
  return cs->session_id;
}

void Master::prepare()
{
	debug(SECTION_MASTER, 10, "---> prepare\n");
	uint32_t mem_size = 1 << 30;
	mem = new char[mem_size];
	mm = new MemoryManager(mem, mem_size, 256 << 10); 
	mm->Init();

	debug(SECTION_MASTER, 10, "init client rpc\n");
	rpc_client = new RdmaRpc(*mm);
	rpc_client->Init(RPC_RDMA | RPC_SHARED_RECV_QUEUE);

	debug(SECTION_MASTER, 10, "init server rpc\n");
	rpc_server = new RdmaRpc(*mm);
	rpc_server->Init(RPC_SERVER | RPC_RDMA | RPC_SHARED_RECV_QUEUE);
	rpc_server->BindSess(uri.c_str());
	debug(SECTION_MASTER, 10, "register file_op_service\n");
	rpc_server->RpcServiceRegister(M_CHUNKOP_SERVICE, chunk_op_service);
	rpc_server->RpcServiceRegister(M_SYSSTATUS_SERVICE, sys_status_service);
	rpc_server->RpcServiceRegister(M_HEARTBEAT_SERVICE, heartbeat_service);

	rpc_server->RpcServiceProviderRegister(M_CHUNKOP_SERVICE, this);
	rpc_server->RpcServiceProviderRegister(M_SYSSTATUS_SERVICE, this);
	rpc_server->RpcServiceProviderRegister(M_HEARTBEAT_SERVICE, this);
	state = ACTIVE;
}

void Master::exit()
{
	debug(SECTION_MASTER, 10, "---> exit\n");
	//TODO close rpc, free memory... 
	debug(SECTION_MASTER, 10, "<--- exit\n");
}


